package com.sitech.ibnms.c3p.util;

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

import org.apache.ignite.IgniteLogger;
import org.apache.ignite.Ignition;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.resources.LoggerResource;
import org.apache.ignite.spi.IgniteSpiConfiguration;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinderAdapter;

import javax.sql.DataSource;
import java.net.InetSocketAddress;
import java.sql.*;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

public class TcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
    public static final String GET_ADDRS_QRY = "select hostname, port from tbl_addrs_ignite where grid_name = ? and port > 0 and last_checked_at > ?";
    public static final String REG_ADDR_QRY = "insert into tbl_addrs_ignite values (?, ?, ?, ?)";
    public static final String UNREG_ADDR_QRY = "delete from tbl_addrs_ignite where hostname = ? and port = ? and grid_name = ?";
    public static final String UNREG_ALL_ADDR_QRY = "delete from tbl_addrs_ignite where grid_name = ?";
    public static final String CREATE_ADDRS_TABLE_QRY = "create table tbl_addrs_ignite (hostname VARCHAR(1024), port INT, grid_name VARCHAR(1024), last_checked_at TIMESTAMP)";
    public static final String CHK_QRY = "select count(*) from tbl_addrs_ignite";
    @LoggerResource
    private IgniteLogger log;
    private DataSource dataSrc;
    private boolean initSchema = true;
    @GridToStringExclude
    private final AtomicBoolean initGuard = new AtomicBoolean();
    @GridToStringExclude
    private final CountDownLatch initLatch = new CountDownLatch(1);

    public TcpDiscoveryJdbcIpFinder() {
        this.setShared(true);
    }

    public Collection<InetSocketAddress> getRegisteredAddresses() throws IgniteSpiException {
        this.init();
        Connection conn = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;

        try {
            conn = this.dataSrc.getConnection();
            stmt = conn.prepareStatement(GET_ADDRS_QRY);
            stmt.setString(1, ignite.name());
            stmt.setTimestamp(2, new Timestamp(System.currentTimeMillis() - 1000 * 100));
            rs = stmt.executeQuery();
            LinkedList<InetSocketAddress> e = new LinkedList<InetSocketAddress>();

            while (rs.next()) {
                e.add(new InetSocketAddress(rs.getString(1), rs.getInt(2)));
            }
            return e;
        } catch (SQLException var9) {
            throw new IgniteSpiException("Failed to get registered addresses version.", var9);
        } finally {
            U.closeQuiet(rs);
            U.closeQuiet(stmt);
            U.closeQuiet(conn);
        }
    }

    public void registerAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
        assert !F.isEmpty(addrs);

        this.init();
        Connection conn = null;
        PreparedStatement stmtReg = null;
        PreparedStatement stmtUnregAll = null;
        boolean committed = false;
        LinkedHashSet<InetSocketAddress> addresses = new LinkedHashSet<InetSocketAddress>(addrs);
        addresses.addAll(getRegisteredAddresses());
        try {
            conn = this.dataSrc.getConnection();
            conn.setAutoCommit(false);
//            conn.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
            stmtReg = conn.prepareStatement(REG_ADDR_QRY);

            try {
                if (ignite.cluster().topologyVersion() > 0) {
                    for (ClusterNode node : ignite.cluster().nodes()) {
                        if (node instanceof TcpDiscoveryNode) {
                            addresses.addAll(((TcpDiscoveryNode) node).socketAddresses());
                        }
                    }
                    // 定期检查时, 先清理旧数据(1分钟超时)
                    stmtUnregAll = conn.prepareStatement(UNREG_ALL_ADDR_QRY);
                    stmtUnregAll.setString(1, ignite.name());
                    stmtUnregAll.execute();
                    stmtUnregAll.close();
                }
            } catch (Exception e) {
                log.error("Failed to cleanUp!", e);
            }
            Timestamp lastCheckedAt = new Timestamp(System.currentTimeMillis());
            for (InetSocketAddress addr : addresses) {
                if (!addr.getAddress().isLoopbackAddress()) { // 端口为0时或本地回环地址(127.0.0.1)不插入数据库
                    stmtReg.setString(1, addr.getAddress().getHostAddress());
                    stmtReg.setInt(2, addr.getPort());
                    stmtReg.setString(3, ignite.name());
                    stmtReg.setTimestamp(4, lastCheckedAt);
                    stmtReg.addBatch();
                }
            }

            stmtReg.executeBatch();
            stmtReg.close();
            conn.commit();
            committed = true;
        } catch (SQLException var11) {
            U.rollbackConnectionQuiet(conn);
            throw new IgniteSpiException("Failed to register addresses: " + addrs, var11);
        } finally {
            if (!committed) {
                U.rollbackConnectionQuiet(conn);
            }

            U.closeQuiet(stmtReg);
            U.closeQuiet(stmtUnregAll);
            U.closeQuiet(conn);
        }
    }

    public void unregisterAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
        assert !F.isEmpty(addrs);

        this.init();
        Connection conn = null;
        PreparedStatement stmt = null;
        boolean committed = false;

        try {
            conn = this.dataSrc.getConnection();
            conn.setAutoCommit(false);
//            conn.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
            stmt = conn.prepareStatement(UNREG_ADDR_QRY);

            for (InetSocketAddress addr : addrs) {
                stmt.setString(1, addr.getAddress().getHostAddress());
                stmt.setInt(2, addr.getPort());
                stmt.setString(3, ignite.name());
                stmt.addBatch();
            }

            stmt.executeBatch();
            conn.commit();
            committed = true;
        } catch (SQLException var10) {
            U.rollbackConnectionQuiet(conn);
            throw new IgniteSpiException("Failed to unregister addresses: " + addrs, var10);
        } finally {
            if (!committed) {
                U.rollbackConnectionQuiet(conn);
            }

            U.closeQuiet(stmt);
            U.closeQuiet(conn);
        }
    }

    @IgniteSpiConfiguration(
            optional = false
    )
    public void setDataSource(DataSource dataSrc) {
        this.dataSrc = dataSrc;
    }

    @IgniteSpiConfiguration(
            optional = true
    )
    public void setInitSchema(boolean initSchema) {
        this.initSchema = initSchema;
    }

    private void init() throws IgniteSpiException {
        if (this.initGuard.compareAndSet(false, true)) {
            if (this.dataSrc == null) {
                throw new IgniteSpiException("Data source is null (you must configure it via setDataSource(..) configuration property)");
            }

            if (!this.initSchema) {
                this.initLatch.countDown();
                this.checkSchema();
                return;
            }

            Connection conn = null;
            Statement stmt = null;
            boolean committed = false;

            try {
                conn = this.dataSrc.getConnection();
                conn.setAutoCommit(false);
//                conn.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
                try {
                    stmt = conn.createStatement();
                    stmt.execute(CHK_QRY);
                } catch (SQLException e) {
                    stmt = conn.createStatement();
                    stmt.executeUpdate(CREATE_ADDRS_TABLE_QRY);
                }
                conn.commit();
                committed = true;
                if (this.log.isDebugEnabled()) {
                    this.log.debug("DB schema has been initialized.");
                }
            } catch (SQLException var8) {
                U.rollbackConnectionQuiet(conn);
                log.error("Failed to initialize DB schema.", var8);
                Ignition.kill(true);
            } finally {
                if (!committed) {
                    U.rollbackConnectionQuiet(conn);
                }

                U.closeQuiet(stmt);
                U.closeQuiet(conn);
                this.initLatch.countDown();
            }
        } else {
            this.checkSchema();
        }

    }

    private void checkSchema() throws IgniteSpiException {
        try {
            U.await(this.initLatch);
        } catch (IgniteInterruptedCheckedException var10) {
            throw new IgniteSpiException("Thread has been interrupted.", var10);
        }

        Connection conn = null;
        Statement stmt = null;

        try {
            conn = this.dataSrc.getConnection();
            stmt = conn.createStatement();
            stmt.execute(CHK_QRY);
        } catch (SQLException var8) {
            throw new IgniteSpiException("IP finder has not been properly initialized.", var8);
        } finally {
            U.closeQuiet(stmt);
            U.closeQuiet(conn);
        }

    }

    public String toString() {
        return S.toString(TcpDiscoveryJdbcIpFinder.class, this);
    }
}

